package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{Column, DataFrame, SparkSession}

object Demo7BurksTest1 {
  def main(args: Array[String]): Unit = {

    //创建SparkSession对象
    /**
     * 在新版本的spark中，如果想要编写spark sql的话，需要使用新的spark入口类：SparkSession
     */
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local")
      .appName("公司营收额数据需求演示")
      .config("spark.sql.shuffle.partitions", 1) //默认分区的数量是200个
      .getOrCreate()

    //导入Spark sql中所有的sql隐式转换函数
    import org.apache.spark.sql.functions._
    //导入另一个隐式转换，后面可以直接使用$函数引用字段进行处理,如果需要做RDD和DF之间的转换
    import sparkSession.implicits._

    //读取数据
    val burksDF: DataFrame = sparkSession.read
      .format("csv")
      .option("sep", ",")
      .schema("burk STRING,year STRING" +
        ",tsl01 DOUBLE,tsl02 DOUBLE,tsl03 DOUBLE,tsl04 DOUBLE" +
        ",tsl05 DOUBLE,tsl06 DOUBLE,tsl07 DOUBLE,tsl08 DOUBLE" +
        ",tsl09 DOUBLE,tsl10 DOUBLE,tsl11 DOUBLE,tsl12 DOUBLE")
      .load("spark/data/burks.txt")

    /**
     * 1、统计每个公司每年按月累计收入  行转列 --> sum窗口函数
     * 输出结果
     * 公司代码,年度,月份,当月收入,累计收入
     *
     */
    //纯sql的方式实现
//    burksDF.createOrReplaceTempView("burks")
//    sparkSession.sql(
//      """
//        |select
//        |t1.burk as burk,
//        |t1.year as year,
//        |t1.month as month,
//        |t1.tsl as tsl,
//        |sum(t1.tsl) over(partition by t1.burk,t1.year order by t1.month) as leijia
//        |from
//        |(select
//        |  burk,
//        |  year,
//        |  month,
//        |  tsl
//        | from
//        |  burks
//        |  lateral view explode(map(1,tsl01,2,tsl02,3,tsl03,4,tsl04,5,tsl05,6,tsl06,7,tsl07,8,tsl08,9,tsl09,10,tsl10,11,tsl11,12,tsl12)) T as month,tsl
//        |) t1
//        |""".stripMargin).show()

    // 使用DSL API实现

    val m: Column = map(
      expr("1"), $"tsl01",
      expr("2"), $"tsl02",
      expr("3"), $"tsl03",
      expr("4"), $"tsl04",
      expr("5"), $"tsl05",
      expr("6"), $"tsl06",
      expr("7"), $"tsl07",
      expr("8"), $"tsl08",
      expr("9"), $"tsl09",
      expr("10"), $"tsl10",
      expr("11"), $"tsl11",
      expr("12"), $"tsl12"
    )

    burksDF
      .select($"burk",$"year",explode(m) as Array("month","tsl"))
      //按月累计
      .withColumn("leijia",sum($"tsl") over Window.partitionBy($"burk",$"year").orderBy($"month"))
//      .show()


    /**
     * 2、统计每个公司当月比上年同期增长率  行转列 --> lag窗口函数
     * 公司代码,年度,月度,增长率（当月收入/上年当月收入 - 1）
     *
     * 1001 1 99
     * 1001 1 100 99
     */
    burksDF
      //行专列
      .select($"burk",$"year",explode(m) as Array("month","tsl"))
      //去上一年同期的收入
      .withColumn("last_tsl",lag($"tsl",1) over Window.partitionBy($"burk",$"month").orderBy($"year"))
      //计算增长率
      .withColumn("p",round($"tsl" / $"last_tsl" - 1,5))
      .withColumn("p1",when($"p".isNotNull,$"p").otherwise(1.0))
      .select($"burk",$"year",$"month",$"tsl",$"last_tsl",$"p1")
      .show()

  }
}
